from pyspark.sql import SparkSession
from pyspark.sql.functions import *
from pyspark.sql.types import *

# 定义Schema
schema = StructType([
    StructField("user_id", StringType()),
    StructField("property_id", StringType()),
    StructField("view_time", TimestampType()),
    StructField("duration", IntegerType()),
    StructField("city", StringType()),
    StructField("price_range", StringType()),
    StructField("device", StringType())
])

spark = SparkSession.builder \
    .appName("FixedStreamingApp") \
    .getOrCreate()

# 流读取
ssc = spark.readStream \
    .schema(schema) \
    .option("header", "true") \
    .option("maxFilesPerTrigger", 1) \
    .csv("./rental_views.csv")  # 修改为实际路径

# 4. 数据预处理
parsed_stream = ssc \
    .withColumn("view_time", to_timestamp(col("view_time"))) \
    .withColumn("duration", col("duration").cast("integer")) \
    .withColumn("price_min", split(col("price_range"), "-").getItem(0).cast("integer")) \
    .withColumn("price_max", split(col("price_range"), "-").getItem(1).cast("integer")) \
    .withColumn("view_date", to_date(col("view_time"))) \
    .withColumn("view_hour", hour(col("view_time")))

# 5. 实时分析（窗口聚合+多维度统计）
city_stats = parsed_stream.groupBy(
    window(col("view_time"), "5 minutes"),
    "city"
).agg(
    count("*").alias("view_count"),
    avg("duration").alias("avg_duration"),
    countDistinct("user_id").alias("unique_users")
)

# 6. 输出到控制台
query = city_stats.writeStream \
    .outputMode("complete") \
    .format("console") \
    .option("truncate", "false") \
    .option("checkpointLocation", "checkpoints/") \
    .start()

# 7. 启动流处理
ssc.start()
ssc.awaitTermination()